package com.yujiahao.bigdata.sql

import com.atguigu.bigdata.sql.SparkSQL01_DataFrame_RDD_DataSet.User
import org.apache.spark.sql.{Encoder, _}
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}

object SparkSQL01_UDAF_OldClass {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkSQL直接构建无法构建，包是私有的，使用伴生对象伴生类构建builder
    val spark = SparkSession.builder
      .master("local")
      .appName("Word Count")
      .getOrCreate()
    //TODO 这里跟命令行的操作不一样，要进行隐式转换，将环境保证好
    import spark.implicits._
    //构建DataSet
    val df: DataFrame = spark.read.json("data/user.json")
    val ds: Dataset[User] = df.as[User]
    //TODO 使用老的API进行函数的自定义：UDAF，Spark早期版本，无法将UDAF强类型函数应用在SQL中
    // UDAF强类型函数主要应用于DSL语法,直接使用自定义函数对象
    ds.select(new MyAvgAge().toColumn).show()

    //TODO 2、关闭资源
    spark.stop()
  }

  //Todo 自定义聚合函数（UDAF）（强类型）考虑泛型
  // 1、继承类Aggregator
  // 2、明确泛型：Spark早期版本，无法将UDAF强类型函数应用在SQL中，UDAF强类型函数主要应用于DSL语法
  // IN:因此输入类型是User对象
  // BUFF:都是Long
  // OUT:Long
  // 3、实现6种方法
  // 4、注意：UDAF强类型函数主要应用于DSL语法,直接使用自定义函数对象

  class MyAvgAge extends Aggregator[User,AvgVuffer,Long] {

    //清空缓冲区
    override def zero: AvgVuffer = {
      AvgVuffer(0L,0L)
    }
    //对缓冲区的数据进行聚合操作
    override def reduce(b: AvgVuffer, a: User): AvgVuffer = {
      b.total += a.age
      b.cnt +=1
      b
    }
    //合并缓冲区
    override def merge(b1: AvgVuffer, b2: AvgVuffer): AvgVuffer = {

      b1.total += b2.total
      b1.cnt += b2.cnt
      b1
    }
    //完成计算
    override def finish(reduction: AvgVuffer): Long = {
      reduction.total / reduction.cnt
    }

    override def bufferEncoder: Encoder[AvgVuffer] = Encoders.product

    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }

  //使用样例类将缓冲区里面的类型进行封装
  case class AvgVuffer(var total: Long, var cnt: Long)
  case class User(id:Long,name:String,age:Long)
}
